package com.atguigu.dataStreamApi.transformation;

import com.alibaba.fastjson.JSON;
import com.atguigu.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import util.SourceUtil;

/**
 * @author WEIYUNHUI
 * @date 2024/11/15 16:22
 *
 * 函数类:
 *   1. 普通函数类 , 只关注函数本身对数据处理的功能()
 *
 *          算子             传入函数（抽象方法）         实现函数的方法
 *         map()     ->    MapFunction         ->  map()
 *         filter()  ->    FilterFunction      ->  filter()
 *         flatmap() ->    FlatMapFunction     ->  flatMap()
 *         reduce()  ->    ReduceFunction      ->  reduce
 *         ......
 *
 *   2. 富函数类（功能更加强大和多）
 *   富函数传入的函数（接口）带有Rich前缀，例如map的传入函数RichMapFunction
 *   RichMapFunction抽象方法，继承RichFunction
 *         RichFunction（接口）
 *            AbstractRichFunction
 *                  算子              传入的函数(抽象方法）              函数的方法
 *                  map()     ->    RichMapFunction         ->  map()
 *                  filter()  ->    RichFilterFunction      ->  filter()
 *                  flatmap() ->    RichFlatMapFunction     ->  flatMap()
 *                  reduce()  ->    RichReduceFunction      ->  reduce
 *                  ......
 *
 *   3. 富函数类的功能:
 *        1)  对数据处理的功能，
 *
 *        2)  生命周期方法 , 算子的每个并行实例
 *               open()    算子的每个并行实例创建的时候调用一次 (在数据处理之前)
 *               xxxx()    对数据进行处理的方法
 *               close()   算子的每个并行实例销毁的时候调用一次(在数据处理之后)
 *
 *        3)  运行时上下文对象 RuntimeContext ， 可以获取各种信息 及 获取按键分区状态进行状态编程
 *
 *             状态编程:
 *                 ①  getState()
 *                 ②  getListState()
 *                 ③  getMapState()
 *                 ④  getReducingState()
 *                 ⑤  getAggregatingState()
 *
 */

public class Flink05_RichFunction {
       public static void main(String[] args) {
       //1.创建flink环境
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
               env.setParallelism(1);

       //对接无界数据源
          // DataStreamSource<Event> ds= env.fromSource(SourceUtil.getSource(), WatermarkStrategy.noWatermarks(), "dataGenSource");

           //2.对接数据源，创建文件连接器对象(有界)
           FileSource<String> fileSource = FileSource.forRecordStreamFormat(
                   new TextLineInputFormat(),
                   new Path("F:\\代码\\11_12flink\\FlinkTutorial\\src\\main\\java\\input\\event.txt")
           ).build();
       //3.读取数据
           DataStreamSource<String> filesource
                   = env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "filesource");
           //1.从文件中读数据
           //1.1无界数据量
           //1.2有界数据流

           SingleOutputStreamOperator<Event> ds = filesource.map(
                   line -> {
                       String[] file = line.split(",");
                       return new Event(file[0].trim(), file[1].trim(), Long.valueOf(file[2].trim()));
                   }
           );

           ds.print("INPUT");

           //4.转换数据
           SingleOutputStreamOperator<String> jsonDS = ds.map(
                   new RichMapFunction<Event, String>() {
                       @Override
                       public void open(Configuration parameters) throws Exception {
                           System.out.println("获取jedis对象。。。。");;
                       }

                       @Override
                       public void close() throws Exception {
                           System.out.println("释放jedis。。。。");
                       }

                       @Override
                       public String map(Event event) throws Exception {

                           System.out.println("访问Redis,读写数据.....");
                           return JSON.toJSONString(event);
                       }
                   }
           );

           jsonDS.print("map");


           //4.打印输出
       //5.关闭资源





               try {
                   env.execute();
               } catch (Exception e) {
                   throw new RuntimeException(e);
               }
           }
}

